package com.bml.architect.filter;

import com.bml.architect.utils.MqConstants;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

/**
 * TAG过滤方式有一个很大的限制，就是一个消息只能有一个TAG，这在一些比较复杂的场景就有点不
 * 足了。 这时候，可以使用SQL表达式来对消息进行过滤
 *
 * 这个模式的关键是在消费者端使用MessageSelector.bySql(String sql)返回的一个
 * MessageSelector。这里面的sql语句是按照SQL92标准来执行的。sql中可以使用的参数有默认的
 * TAGS和一个在生产者中加入的a属性。
 * SQL92语法：
 * RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。
 * 数值比较，比如：>，>=，<，<=，BETWEEN，=；
 * 字符比较，比如：=，<>，IN；
 * IS NULL 或者 IS NOT NULL；
 * 逻辑符号 AND，OR，NOT；
 * 常量支持类型为：
 * 数值，比如：123，3.1415；
 * 字符，比如：'abc'，必须用单引号包裹起来；
 * NULL，特殊的常量
 * 布尔值，TRUE 或 FALSE
 * 使用注意：只有推模式的消费者可以使用SQL过滤。拉模式是用不了的
 * 因为过滤是在broker端的，不是消费端的
 */
public class SqlFilterProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("filterProducerGroup");
        producer.setNamesrvAddr(MqConstants.getNameServerAddr());
        producer.start();
        String tags[] = new String[]{"TAGA", "TAGB", "TAGC"};
        for (int i = 0; i < 10; i++) {
            Message message = new Message("TopicTest", tags[i % tags.length], ("rocketMq tag filter" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            message.putUserProperty("num",String.valueOf(i));
            SendResult send = producer.send(message);
            if (send.getSendStatus().equals(SendStatus.SEND_OK)) {
                System.out.println("消息发送成功 " + i);
            }
        }
        producer.shutdown();
    }
}


